package com.intct;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.ObjectPath;

import java.util.Map;

/**
 * @author gufg
 * @since 2025-10-21 09:58
 */
public class OdsDatabaseApp {
    public static void main(String[] args) throws Exception {
        // 获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并度
        env.setParallelism(1);
        // 开启检查
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // SQL环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 配置源--mysql cdc
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("cdh-node")
                .port(13306)
                .databaseList("m1") // 设置捕获的数据库， 如果需要同步整个数据库，请将 tableList 设置为 ".*".
                .tableList("m1.test05") // 设置捕获的表
                .username("root")
                .password("Test_090110")
                .deserializer(new JsonDebeziumDeserializationSchema()) // 将 SourceRecord 转换为 JSON 字符串
                /*启动模式
                    配置选项scan.startup.mode指定 MySQL CDC 使用者的启动模式。有效枚举包括：
                    initial （默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的 binlog。
                    earliest-offset：跳过快照阶段，从可读取的最早 binlog 位点开始读取
                    latest-offset：首次启动时，从不对受监视的数据库表执行快照， 连接器仅从 binlog 的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。
                    specific-offset：跳过快照阶段，从指定的 binlog 位点开始读取。位点可通过 binlog 文件名和位置指定，或者在 GTID 在集群上启用时通过 GTID 集合指定。
                    timestamp：跳过快照阶段，从指定的时间戳开始读取 binlog 事件。
                */
//                .startupOptions(StartupOptions.latest())
                .chunkKeyColumn(new ObjectPath("m1", "test05"), "id")
                .build();

        // 接收数据
        DataStreamSource<String> sourceDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "cdc_ds_app_source");

        SingleOutputStreamOperator<Schema> mapDS = sourceDS.map(new MapFunction<String, Schema>() {
            @Override
            public Schema map(String value) throws Exception {

                JSONObject jsonObj = JSON.parseObject(value);

                Schema.Builder builder = Schema.newBuilder();

                for (Map.Entry<String, Object> entry : jsonObj.getJSONObject("after").entrySet()) {
                    builder.column(entry.getKey(), entry.getValue().toString());
                }

                return builder.build();
            }
        });

//        tenv.createTemporaryView("test", mapDS, schema);
//
//        tenv.executeSql("select * from test").print();

        env.execute();
    }
}
